#利用sql进行数据分析 提交到pyspark集群中
# coding:utf8
#@Author：LU80
from pyhive import hive
import pandas as pd
from sqlalchemy import create_engine
#虚拟机MySQL链接
engine = create_engine('mysql+pymysql://root:123456@192.168.88.161:3306/book_movies_olpk')

if __name__ == '__main__':
    # 获取到Hive(Spark ThriftServer的链接)
    conn = hive.Connection(host="node1", port=10000, username="root", database='book_movies_olpk')
    cursor = conn.cursor()

    #Task-1 每个国家参赛选手数据分析
    cursor.execute("SELECT noc ,count(noc)  FROM olpk group by noc;")
    # 通过fetchall API 获得返回值
    result = cursor.fetchall()
    #list类型的result转为df类型
    df = pd.DataFrame(result)
    df.columns=['noc', 'not_count']
    print(df)
    # 写出df到mysql数据库中
    df.to_sql('olpk_test1', engine, index=False)
    print("Task—1 finished")

    # Task-2 每个国家奖牌总数变化
    cursor.execute("SELECT noc ,count(*) as count_medal  FROM olpk where medal!='NA' group by noc order by count_medal desc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['noc', 'count_medal']
    print(df)
    df.to_sql('olpk_test2', engine, index=False)
    print("Task—2 finished")

    #Task-3  每个国家金牌数量变化数据分析
    cursor.execute("SELECT noc ,count(*) as count_medal  FROM olpk where medal=='Gold' group by noc order by count_medal desc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['noc', 'count_medal_gold']
    print(df)
    df.to_sql('olpk_test3', engine, index=False)
    print("Task—3 finished")

    #Task-4  季节奥运会国家奖牌总数数据分析
    cursor.execute("SELECT season ,noc,count(*) as count_medal_gold  FROM olpk  group by season,noc order by count_medal_gold desc")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['season', 'noc','count_medal_gold' ]
    print(df)
    df.to_sql('olpk_test4', engine, index=False)
    print("Task—4 finished")

    #Task-5  美国历年获得奖牌数
    cursor.execute("SELECT sport ,count(*) as count_medal_usa FROM olpk where noc=='USA' group by sport order by count_medal_usa desc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['sport', 'count_medal_usa']
    print(df)
    df.to_sql('olpk_test5', engine, index=False)
    print("Task—5 finished")

    #Task-6  各个项目男生总数对比
    cursor.execute("SELECT sport ,count(*) as count_medal_m FROM olpk where sex=='M' and medal!='NA' group by sport order by count_medal_m desc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['sport', 'count_medal_m']
    print(df)
    df.to_sql('olpk_test6', engine, index=False)
    print("Task—6 finished")

    #Task-7  各个项目女生总数对比
    cursor.execute("SELECT sport ,count(*) as count_medal_f FROM olpk where sex=='F' and medal!='NA' group by sport order by count_medal_f desc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['sport', 'count_medal_f']
    print(df)
    df.to_sql('olpk_test7', engine, index=False)
    print("Task—7 finished")

    # Task-8  每个国家参赛选手身高数据分析
    cursor.execute("SELECT noc ,avg(height) as avg_height  FROM olpk  group by noc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['noc', 'avg_height']
    print(df)
    df.to_sql('olpk_test8', engine, index=False)
    print("Task—8 finished")

    # Task-9  每个国家参赛选手体重数据分析
    cursor.execute("SELECT noc ,avg(weight) as avg_weight  FROM olpk  group by noc;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['noc', 'avg_weight']
    print(df)
    df.to_sql('olpk_test9', engine, index=False)
    print("Task—9 finished")

    # Task-10  奥运会举办城市产生的奖牌选手
    cursor.execute("SELECT city,count(*) count_medal_gole FROM olpk group by city;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['city', 'count_medal_gole']
    print(df)
    df.to_sql('olpk_test10', engine, index=False)
    print("Task—10 finished")

    #Task-11  用户所有被转发的总数，输出前10个用户
    cursor.execute("select b.id,sum(b.cnt) as bsum "
                   "from(select get_json_object(a.j,'$.userId') as id,get_json_object(a.j,'$.reportCount') as cnt "
                        "from (select substring(json,2,length(json)-2) as j "
                              "from weibo) a) b "
                   "group by b.id "
                   "order by bsum desc "
                   "limit 10;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['id', 'bsum']
    print(df)
    df.to_sql('olpk_test11', engine, index=False)
    print("Task—11 finished")

    #Task-12  被转发次数最多的前10条微博，输出用户id
    cursor.execute("select distinct get_json_object(a.j,'$.userId') as id,cast(get_json_object(a.j,'$.reportCount') as INT) as cnt "
                   "from(select substring(json,2,length(json)-2) as j "
                   "     from weibo) a "
                   "order by cnt desc "
                   "limit 10;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['id', 'cnt']
    print(df)
    df.to_sql('olpk_test12', engine, index=False)
    print("Task—12 finished")

    # Task-13  用户发布的微博总数
    cursor.execute("select get_json_object(a.j,'$.userId'),count(1)"
                   "from(select substring(json,2,length(json)-2) as j "
                        "from weibo) a "
                   "group by get_json_object(a.j,'$.userId') "
                   "limit 50;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['uid', 'sum']
    print(df)
    df.to_sql('olpk_test13', engine, index=False)
    print("Task—13 finished")

    # Task-14  微博中评论次数小于100的用户id和数据来源
    cursor.execute("select distinct get_json_object(a.j,'$.userId') as id,get_json_object(a.j,'$.source') as source "
                   "from(select substring(json,2,length(json)-2) as j from weibo) a "
                   "where get_json_object(a.j,'$.commentCount')<1000 and get_json_object(a.j,'$.source') "
                   "like '%端%' "
                   "limit 50;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['id', 'source']
    print(df)
    df.to_sql('olpk_test14', engine, index=False)
    print("Task—14 finished")

    # Task-15  找出20位发微博的独立用户的数据来源和用户id
    cursor.execute("select distinct get_json_object(a.j,'$.userId') as id, get_json_object(a.j,'$.source') as phone "
                   "from(select substring(json,2,length(json)-2) as j "
                        "from weibo) a "
                   "where lower(get_json_object(a.j,'$.source')) like '%iphone%' "
                   "limit 20;")
    result = cursor.fetchall()
    df = pd.DataFrame(result)
    df.columns = ['id', 'phone']
    print(df)
    df.to_sql('olpk_test15', engine, index=False)
    print("Task—15 finished")